package com.pajk.user.dal.index;

import com.pajk.user.service.common.tair.TairNamespaceEnum;
import com.taobao.tair.DataEntry;
import com.taobao.tair.Result;
import com.taobao.tair.ResultCode;
import com.taobao.tair.TairManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.*;

/**
 * Tair implemented Index
 *
 * Support any key type that implements Serializable interface;
 * Support two kinds of value types:
 * 1. ValueType that is Serializeble;
 * 2. ValueType that is kind of SortedSet;
 *
 * TODO complete failover strategy to ensure eventual consistency
 *
 * @author Haomin Liu
 */
public class TairIndex<KeyType extends Serializable, ValueType> implements Index<KeyType, ValueType> {

    private static final Logger Log = LoggerFactory.getLogger(TairIndex.class);

    public static final int LISTVALUE_MAX_COUNT=200;

    private TairNamespaceEnum namespace;
    private TairManager tairManager;
    private Class<KeyType> keyClass;
    private Class<ValueType> valueClass;
    private GeneralTairKeyValueOps<KeyType, ValueType> tairWrap;


    public TairIndex(TairManager tairManager, TairNamespaceEnum namespace, Class<KeyType> keyClass, Class<ValueType> valueClass){

        if((tairManager == null)||
           (namespace == null)||
           (keyClass == null)||
           (valueClass == null))
        {
            throw new IllegalArgumentException(
                    String.format("constructor args contains null value! tairManager = %s, namespace = %s, keyClass = %s, valueClass=%s",
                                  tairManager, namespace, keyClass, valueClass));
        }

        this.tairManager = tairManager;
        this.namespace = namespace;
        this.keyClass = keyClass;
        this.valueClass = valueClass;

        // check if valueClass is derived from Lists
        if(SortedSet.class.isAssignableFrom(valueClass)){
            tairWrap = new SortedSetValueTairOps<KeyType, ValueType>();
        }else if(Serializable.class.isAssignableFrom(valueClass)){
            tairWrap = new SerializableValueTairOps<KeyType, ValueType>();
        }else{
            throw new IllegalArgumentException(String.format("ValueType class is neither Serializable classes nor SortedSet classes! ValueType = %s", valueClass));
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public ValueType get(KeyType key){
        if((key == null))
        {
            throw new IllegalArgumentException(
                    String.format("input args contains null value! key=%s", key));
        }

        try{
            DataEntry res = doGet(key);
            return (res == null) ? null : (ValueType)res.getValue();
        }catch(RuntimeException re){
            Log.error("RuntimeException catched.", re);
            return null;
        }catch(Throwable t){
            Log.error("Throwable catched.", t);
            return null;
        }
    }

    @Override
    public void put(KeyType key, ValueType val){
        if((key == null)||(val == null))
        {
            throw new IllegalArgumentException(
                    String.format("input args contains null value! kay = %s, val = %s", key, val));
        }

        try{
            // try get data version
            DataEntry res = doGet(key);
            if (res == null){
                // KV not exist for now, insert new entry into tairindex
                ResultCode rc = doInsert(key, val);
                if(!rc.isSuccess()){
                    RuntimeException e = new RuntimeException(String.format("Insert data(key=%s, val=%s) error! Tair result code: %s", key, val, rc));
                    throw e;

                }
                return;
            }

            // key already exists, do optimistic put, try 3 times
            int nTries = 3;
            Log.debug("doCasPut(this={}, res={}, key={}, val={}, nTries={})", this, res, key, val, nTries);
            if(false == doCasPut(key, val, res.getVersion(), nTries)){
                // still failed, send retry message, do this task in a async manner
                // stackAsyncTask(AsyncIndexTask.ACTION_PUT, key, val);
                Log.error("Put KV into tairindex failed, tried {} times.  ", nTries);
            }
        }catch(RuntimeException re){
            Log.error("RuntimeException catched.", re);
        }catch(Throwable t){
            Log.error("RuntimeException catched.", t);
        }
    }

    @Override
    public void remove(KeyType key){
        if((key == null))
        {
            throw new IllegalArgumentException(
                    String.format("input args contains null value! key = %s", key));
        }

        try{
            doRemove(key, null);
        }catch(RuntimeException re){
            Log.error("RuntimeException catched.", re);
        }catch(Throwable t){
            Log.error("RuntimeException catched.", t);
        }
    }

    @Override
    public void remove(KeyType key, Object subKey){
        if((key == null)||(subKey == null))
        {
            throw new IllegalArgumentException(
                    String.format("input args contains null value! key = %s, subKey = %s", key, subKey));
        }

        try{
            doRemove(key, subKey);
        }catch(RuntimeException re){
            Log.error("RuntimeException catched.", re);
        }catch(Throwable t){
            Log.error("RuntimeException catched.", t);
        }
    }

    public TairNamespaceEnum getNamespace() {
        return namespace;
    }

    public TairManager getTairManager() {
        return tairManager;
    }

    public Class<KeyType> getKeyClass() {
        return keyClass;
    }

    public Class<ValueType> getValueClass() {
        return valueClass;
    }

    private DataEntry doGet(KeyType key){
        Result<DataEntry> result = tairWrap.get(namespace.ordinal(), key);
        if(!result.isSuccess()){
            ResultCode rc = result.getRc();
            if(rc != ResultCode.DATANOTEXSITS){
                Log.warn("Get KV error due to tairindex error: {}", rc.toString());
            }
            return null;
        }
        return result.getValue();
    }

    private ResultCode doInsert(KeyType key, ValueType value){
        return tairWrap.put(namespace.ordinal(), key, value, 0);
    }

    private void doRemove(KeyType key, Object subKey) {
        ResultCode res = tairWrap.remove(namespace.ordinal(), key, subKey);
        if(!res.isSuccess()){
            // stackAsyncTask(AsyncIndexTask.ACTION_REMOVE, key, null);
        }
    }

    private Boolean doCasPut(KeyType key, ValueType value, int version, int retryTimes){
        int ver = version;
        for(int i = 0; i < retryTimes; i++){
            ResultCode code = tairWrap.put(namespace.ordinal(), key, value, ver);
            if (code == ResultCode.SUCCESS){
                return true;
            }

            // get new version
            DataEntry res = doGet(key);
            if(res != null){
                ver = res.getVersion();
            }
        }
        return false;
    }

    private abstract class GeneralTairKeyValueOps<K, V>{

        public abstract ResultCode put(int namespace, K key, V value, int version);

        public abstract Result<DataEntry> get(int namespace, K key);

        public abstract ResultCode remove(int namespace, K key, Object subKey);

    }

    private class SerializableValueTairOps<K extends Serializable, V> extends GeneralTairKeyValueOps<K, V>{

        @Override
        public ResultCode put(int namespace, K key, V value, int version) {
            return tairManager.put(namespace, key, (Serializable) value);
        }

        @Override
        public Result<DataEntry> get(int namespace, K key) {
            return tairManager.get(namespace, key);
        }

        @Override
        public ResultCode remove(int namespace, K key, Object subKey) {
            return tairManager.delete(namespace, key);
        }
    }

    private class SortedSetValueTairOps<K extends Serializable, V> extends GeneralTairKeyValueOps<K, V>{

        @Override
        public ResultCode put(int namespace, K key, V value, int version) {
            List<?> param = new LinkedList((SortedSet<?>) value);

            //ResultCode addItems(int namespace, Serializable key, List<? extends Object> items, int maxCount, int version, int expireTime);
            return tairManager.addItems(namespace, key, param, LISTVALUE_MAX_COUNT, version, 0);
        }

        @Override
        public Result<DataEntry> get(int namespace, K key) {
            // try best to convert return value from List to SortedSet
            Result<DataEntry> res = tairManager.getItems(namespace, key, 0, LISTVALUE_MAX_COUNT);
            if(res != null){
                DataEntry entry = res.getValue();
                if(entry != null){
                    SortedSet<?> set = new TreeSet<Object>((Collection)entry.getValue());
                    res.getValue().setValue(set);
                }
            }

            return res;
        }

        @Override
        public ResultCode remove(int namespace, K key, Object subKey) {
            //	ResultCode removeItems(int namespace, Serializable key, int offset, int count);
            Result<DataEntry> res = tairManager.getItems(namespace, key, 0, LISTVALUE_MAX_COUNT);
            // no entry at all, kind of remove success
            if(res == null){
                return ResultCode.SUCCESS;
            }

            if(res.getRc() == ResultCode.DATANOTEXSITS){
                return ResultCode.SUCCESS;
            }

            // try get the original list and determine to-be-delete item position in the original list, then delete it;
            if(res.isSuccess()){

                if(subKey == null){
                    // "subKey == null" means remove all
                    return tairManager.removeItems(namespace, key, 0, LISTVALUE_MAX_COUNT);
                }

                DataEntry data = res.getValue();
                if(data != null){
                    List origList = (List)data.getValue();
                    SortedSet origSortedSet = new TreeSet(origList);

                    // does not contain this subKey in the first place, return success
                    if(!origSortedSet.contains(subKey)){
                        return ResultCode.SUCCESS;
                    }

                    int subKeyOffset = origSortedSet.headSet(subKey).size() + 1;

                    //	ResultCode removeItems(int namespace, Serializable key, int offset, int count);
                    return tairManager.removeItems(namespace, key, subKeyOffset, 1);
                }
            }
            return res.getRc();
        }
    }
}
